<?php
declare (strict_types=1);

namespace app\command;

use app\model\Data;
use PhpAmqpLib\Exception\AMQPProtocolChannelException;
use think\console\Command;
use think\console\Input;
use think\console\input\Argument;
use think\console\input\Option;
use think\console\Output;
use think\facade\Log;
use Workerman\Lib\Timer;
use Workerman\Worker;

class ConsumerDeviceQueue extends Command
{
    public function consume($message)
    {
        try {
            $data = json_decode($message->body, true);
            $bodyData = $data['data'] ?? null;
            $receiveTime = $data['receive_time'] ?? null;
            if (!$bodyData) {
                throw new \Exception('错误数据，从队列中删除');
            }

            createFrozenDataTable();
            $timestamp = time();
            $serialNumber = pack('H*', substr($bodyData, 0, 26));
            $result = parseData(substr($bodyData, 26), $serialNumber);
            if (!is_array($result) || empty($result)) {
                throw new \Exception('错误数据，从队列中删除');
            }

            $result['serial_number'] = $serialNumber;
            $result['original'] = $bodyData;
            $result['receive_time'] = date('Y-m-d H:i:s', $receiveTime);
            $result['create_time'] = date('Y-m-d H:i:s', $timestamp);
            $result['update_time'] = date('Y-m-d H:i:s', $timestamp);

            $date = date('Y_m', time());
            $tableName = "ice_data_{$date}";
            $model = app(Data::class, [], true)->table($tableName);
            $model->save($result);

            $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
        } catch (\Exception|AMQPProtocolChannelException $e) {
            $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
            Log::record('[设备消费队列出错]' . $e->getMessage() . ';文件:' . $e->getFile() . ';位于:' . $e->getLine(), 'debug');
        }
    }

    protected function configure()
    {
        // 指令配置
        $this->setName('consumer_device_queue')
            ->addArgument('status', Argument::REQUIRED, 'start/stop/reload/status/connections')
            ->addOption('d', null, Option::VALUE_NONE, 'daemon（守护进程）方式启动')
            ->setDescription('监听移动侧设备rabbitmq消息队列进行同步设备数据');
    }

    protected function execute(Input $input, Output $output)
    {
        $this->init($input, $output);
        $this->consumer($input, $output);

        try {
            Worker::runAll();
        } catch (\Exception|\Throwable $e) {
            $output->warning($e->getMessage());
        }
    }

    protected function init(Input $input, Output $output)
    {
        global $argv;
        $argv[1] = $input->getArgument('status') ?? 'start';
        if ($input->hasOption('d')) {
            $argv[2] = '-d';
        } else {
            unset($argv[2]);
        }
    }

    protected function consumer(Input $input, Output $output)
    {
        try {
            $worker = new Worker();
            $worker->name = 'ice_2025_consumer_device_queue';
            $worker::$logFile = '/dev/null';
            $worker->count = 1;
            $worker::$pidFile = runtime_path() . 'consumer_device_queue.pid';
            $worker->onWorkerStart = function () use ($output) {
                $queueName = env('rabbitmq.mq_queue_name');
                $channel = rabbitmqConnect();
                $channel->basic_consume($queueName, '', false, false, false, false, [$this, 'consume']);

                Timer::add(0.01, function () use ($channel, $output) {
                    if (count($channel->callbacks)) $channel->wait();
                });
            };

        } catch (\Exception $e) {
            echo $e->getMessage() . PHP_EOL;
        }
    }
}
